/*
 * PsRecordCasterChannel.cpp
 *
 *  Created on: 2018年9月1日
 *      Author: zhengboyuan
 */

#include "PsRecordCasterChannel.h"
#include "MFormatUtil.h"
#include "CLog.h"
#include "BasicMacro.h"

#include "PsMediaReader.h"
#include "FfmpegUtil.h"
#include "TimeHelper.h"


extern "C"
{
	#include <libavutil/time.h>
}

static const int MAX_SYNC_TIME = 2 * 1000 * 1000;
static const AVRational PREREAD_SCALE = av_make_q(5,2);


namespace av
{

PsRecordCasterChannel::PsRecordCasterChannel():
		m_id(),
		m_sink(),
		m_caster(-1),
		m_index(),
		m_state(av::STATE_STOPPED),
		m_scale(av_make_q(1,1)),
		m_curTime()
{
}

PsRecordCasterChannel::~PsRecordCasterChannel()
{
	close();
}

void PsRecordCasterChannel::setID(int id)
{
	m_id = id;
}

bool PsRecordCasterChannel::open(const PsRecordCasterTransport& transport, MRecord* records, int size, int64_t offset, double scale)
{
	if (!records || size <= 0)
	{
		return false;
	}

	m_records.assign(records, records + size);

	m_reader.reset(new PsMediaReader(records, size));

	int rc = m_reader->open("", "");
	if (rc != 0)
	{
		CLog::warning("failed to open reader.\n");
		return false;
	}

	MediaFormat format;
	m_reader->getFormat(format);
	MFormat fmt = MFormat();
	MFormatUtil::copyFrom(&fmt, format);

	PsCasterParams params = PsCasterParams();
	params.payload = transport.payload;
	params.ssrc = transport.ssrc;
	params.es = transport.esMode;
	rc = pscaster_openEx(&m_caster, transport.protocol, transport.outPort, &fmt, &params);
	if (rc != 0)
	{
		CLog::warning("pscaster_openEx failed. rc: %d\n", rc);
		return false;
	}

	AVRational r = av_d2q(scale, MAX_CASTER_SCALE);
	handleSetScale(r);

	start();

	return true;
}

void PsRecordCasterChannel::close()
{
	if (isRunning())
	{
		stop();
	}

	m_reader->close();
}

bool PsRecordCasterChannel::addTarget(const char* ip, int port)
{

	if (m_caster < 0)
	{
		return false;
	}

	int rc = pscaster_add_target(m_caster, ip, port);

	return rc == 0;
}

bool PsRecordCasterChannel::removeTarget(const char* ip, int port)
{
	if (m_caster < 0)
	{
		return false;
	}

	int rc = pscaster_remove_target(m_caster, ip, port);

	return rc == 0;
}

void PsRecordCasterChannel::removeTargets()
{
	if (m_caster < 0)
	{
		return ;
	}

	pscaster_remove_all_target(m_caster);
}

bool PsRecordCasterChannel::setScale(double scale)
{
	if (scale < MIN_CASTER_SCALE)
	{
		return false;
	}

	if (scale > MAX_CASTER_SCALE)
	{
		return false;
	}

	AVRational r = av_d2q(scale, MAX_CASTER_SCALE);

	postCommand(kSetScale, r.num, r.den);

	return true;
}

bool PsRecordCasterChannel::seek(int64_t offset)
{
	postCommand(kSeek, offset);
	return true;
}

bool PsRecordCasterChannel::startStream()
{
	postCommand(kPlay);

	return true;
}

void PsRecordCasterChannel::stopStream()
{
	postCommand(kStop);
}

void PsRecordCasterChannel::pauseStream()
{
	postCommand(kPause);
}

void PsRecordCasterChannel::setSink(PsRecordCasterSink* sink)
{
	m_sink = sink;
}

void PsRecordCasterChannel::onCommand(Command& cmd)
{
	if (cmd.type == kPlay)
	{
		handlePlay();
	}
	else if (cmd.type == kPause)
	{
		handlePause();
	}
	else if (cmd.type == kStop)
	{
		handleStop();
	}
	else if (cmd.type == kSeek)
	{
		int64_t offset = cmd.at(0).toLong();
		handleSeek(offset);
	}
	else if (cmd.type == kSetScale)
	{
		AVRational scale;
		scale.num = cmd.at(0).toInt();
		scale.den = cmd.at(1).toInt();

		handleSetScale(scale);

		fireMediaEvent(CASTER_EVENT_SCALE, 0);
	}
	else if (cmd.type == kFlush)
	{
		handleFlush();
	}
}

void PsRecordCasterChannel::onIdle()
{
	if (!m_reader)
	{
		timedwait(-1);
		return;
	}

	if (m_state != STATE_PLAYING)
	{
		timedwait(-1);
		return;
	}

	AVPacketPtr pkt(new AVPacket(), ffmpeg::PacketDeleter());
	av_init_packet(pkt.get());
	int rc = m_reader->read(*pkt);
	if (rc == 0)
	{
		int delay = computeDelay(pkt);
		if (delay > 0)
		{
			timedwait(delay);
		}

		dispatchPacket(pkt);
	}
	else if (rc == AVERROR(EAGAIN))
	{
		// continue;
		timedwait(50);
	}
	else
	{
		fireMediaEvent(CASTER_EVENT_END, 0);
		timedwait(-1);
	}
}

void PsRecordCasterChannel::sendPacket(AVPacket& pkt)
{
	//CLog::info("PsRecordCasterChannel::sendPacket. caster: %d, type: %d, size: %d, pts: %lld\n", m_caster, pkt.stream_index, pkt.size, pkt.pts);

	MPacket mpkt = MPacket();
	mpkt.data = pkt.data;
	mpkt.size = pkt.size;
	mpkt.type = pkt.stream_index;
	mpkt.flags = pkt.flags;
	mpkt.pts = pkt.pts;
	mpkt.duration = pkt.duration;

	pscaster_write(m_caster, &mpkt);
}

void PsRecordCasterChannel::fireMediaEvent(int event, int value)
{
	if (m_sink)
	{
		m_sink->onCasterEvent(m_id, event);
	}
}

void PsRecordCasterChannel::handlePlay()
{
    m_reader->play();

	setState(STATE_PLAYING);

	fireMediaEvent(CASTER_EVENT_PLAYING, 0);
}

void PsRecordCasterChannel::handlePause()
{
    if (!m_reader)
    {
        return;
    }

	setState(STATE_PAUSED);

	m_reader->pause();

	fireMediaEvent(CASTER_EVENT_PAUSED, 0);
}

void PsRecordCasterChannel::handleStop()
{
	setState(STATE_STOPPED);

    if (!m_reader)
    {
        return;
    }

    m_reader->stop();

	fireMediaEvent(CASTER_EVENT_STOPPED, 0);

	for (size_t i = 0; i < ARRAY_SIZE(m_lastPoint); i ++)
	{
		m_lastPoint[i].reset();
	}

	m_scale = av_make_q(1,1);
    m_curTime = 0;
}

void PsRecordCasterChannel::handleSeek(int64_t offset)
{
    if (!m_reader)
    {
        return;
    }

	fireMediaEvent(CASTER_EVENT_SEEKING, 0);

	int rc = m_reader->seek(offset);
	if (rc != 0)
	{
		CLog::warning("failed to seek. offset:%d", (int)offset);
	}

	fireMediaEvent(CASTER_EVENT_SEEKED, 0);

	handleFlush();

}

void PsRecordCasterChannel::handleSetScale(AVRational scale)
{
	for (size_t i = 0; i < ARRAY_SIZE(m_lastPoint); i ++)
	{
		m_lastPoint[i].reset();
	}

	m_scale = scale;

}

void PsRecordCasterChannel::handleFlush()
{
	//
}

void PsRecordCasterChannel::setState(StreamState state)
{
	comn::AutoCritSec lock(m_cs);
	m_state = state;
}

int PsRecordCasterChannel::computeDelay(AVPacketPtr& pkt)
{
	int64_t delay = 0;
	int64_t curClock = getClock();
	TimePoint curPoint(curClock, pkt->pts);
	TimePoint& lastPoint = m_lastPoint[pkt->stream_index];

	if (pkt->stream_index != 0)
	{
		delay = 0;
	}

	if (!lastPoint.isSet())
	{
		lastPoint = curPoint;
	}
	else
	{
		int64_t ptsDelta = curPoint.m_pts - lastPoint.m_pts;
		int64_t clockDelta = curPoint.m_clock - lastPoint.m_clock;
		if (ptsDelta > MAX_SYNC_TIME)
		{
			lastPoint.reset();
		}
		else if (ptsDelta < 0)
		{
			lastPoint.reset();
		}
		else
		{
            int64_t pretime = av_rescale_q(ptsDelta, PREREAD_SCALE, m_scale);
            pretime = 0;
			delay = (ptsDelta - clockDelta) - pretime;

			lastPoint.reset(lastPoint.m_clock + ptsDelta, curPoint.m_pts);
		}
	}

	//const char* name = (pkt->stream_index == 0) ? "video" : "audio";
	//CLog::debug("computeDelay. packet(%s) clock:%I64d, pts:%I64d, delay:%d\n", name, curPoint.m_clock/1000, curPoint.m_pts/1000, delay/1000);

	return (int)delay/1000;
}

int64_t PsRecordCasterChannel::getClock()
{
    int64_t clk = util::TimeHelper::getClock();
    if ((m_scale.num > 0) && (m_scale.num != m_scale.den))
    {
        clk = av_rescale(clk, m_scale.num, m_scale.den);
    }
    return clk;
}

void PsRecordCasterChannel::dispatchPacket(AVPacketPtr& pkt)
{
	sendPacket(*pkt);
}



} /* namespace av */
